package com.flink.api.unit6;

import com.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * udf_1
 *
 * @author caizhiyang
 * @since 2024-02-23
 */
public class udf_3 {

    public static void main(String[] args) throws Exception {
        //实现richMapFunction函数
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4);

        stream.map(new RichMapFunction<Integer, Integer>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");
            }

            @Override
            public Integer map(Integer integer) throws Exception {
                return integer + 1;
            }

            @Override
            public void close() throws Exception {
                super.close();
                System.out.println("索引是：" + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
            }
        }).print();

        env.execute();
    }
}


